package comp6231.project.replica;

import java.io.IOException;
import java.math.BigInteger;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.nio.ByteBuffer;
import java.util.Hashtable;

import comp6231.project.common.*;

public abstract class Replica extends Thread {
	
	volatile boolean stop = false;
	protected BigInteger sequenceNumber = BigInteger.valueOf(1);
	protected Hashtable<BigInteger, Message> holdBack = new Hashtable<BigInteger, Message>();
	protected String replicaId;
	
	

	abstract public int buyOp(String customerID, String itemID, int numberOfItem);
	
	abstract public int returnOp(String customerID, String itemID, int numberOfItem);
	
	abstract public int checkStockOp(String itemID);
	
	abstract public int exchangeOp(String customerID, String boughtItemID, int boughtNumber, String desiredItemID, int desiredNumber);
	
	abstract public ReplicaState getReplicaState();
	
	abstract public void setReplicaState(ReplicaState state);
	
	private String getMulticastGroup() {
		if (getReplicaId().equals(Constants.REPLICA_ID_M1) ||
			getReplicaId().equals(Constants.REPLICA_ID_M2) ||
			getReplicaId().equals(Constants.REPLICA_ID_M3)) {
			return Constants.RM_MULTICAST_GROUP_MONTREAL;
			
		} else if (getReplicaId().equals(Constants.REPLICA_ID_T1) ||
				   getReplicaId().equals(Constants.REPLICA_ID_T2) ||
				   getReplicaId().equals(Constants.REPLICA_ID_T3)) {
			return Constants.RM_MULTICAST_GROUP_TORONTO;
			
		} else {
			return Constants.RM_MULTICAST_GROUP_VANCOUVER;
		}
	}
	
	private int getRMPort() {
		if (getReplicaId().equals(Constants.REPLICA_ID_M1) ||
				getReplicaId().equals(Constants.REPLICA_ID_M2) ||
				getReplicaId().equals(Constants.REPLICA_ID_M3)) {
				return Constants.RM_PORT_MONTREAL;
				
			} else if (getReplicaId().equals(Constants.REPLICA_ID_T1) ||
					   getReplicaId().equals(Constants.REPLICA_ID_T2) ||
					   getReplicaId().equals(Constants.REPLICA_ID_T3)) {
				return Constants.RM_PORT_TORONTO;
				
			} else {
				return Constants.RM_PORT_VANCOUVER;
			}		
	}
	
	public Message process(Message msg) {
		try {
			int result = 0;
			if (msg.request.equals(Message.REQUEST_BUY)) {
				result = buyOp(msg.customerID, msg.itemID, msg.numItems);
				
			} else if (msg.request.equals(Message.REQUEST_RETURN)) {
				result = returnOp(msg.customerID, msg.itemID, msg.numItems);
				
			} else if (msg.request.equals(Message.REQUEST_CHECKSTOCK)) {
				result = checkStockOp(msg.itemID);
			
			} else if (msg.request.equals(Message.REQUEST_EXCHANGE)) {
				result = exchangeOp(msg.customerID, msg.itemID, msg.numItems, msg.exchangeItemID, msg.numExchangeItems);
			}
			msg.result = result;
			msg.replicaID = getReplicaId();
			
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		return msg;
	}
	
	public void run() {
		MulticastSocket msIn = null;
		InetAddress group = null;
		DatagramSocket dsOut = null;
		try {
			String groupName = getMulticastGroup();
			group = InetAddress.getByName(groupName);
			msIn = new MulticastSocket(Constants.RM_MULTICAST_PORT);
			msIn.joinGroup(group);

			dsOut = new DatagramSocket();
			
			InetAddress serverIp = null;
			int port = 0;
			

			while (!stop) {
				
				byte bufIn[] = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
				byte bufOut[] = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
				DatagramPacket pack = new DatagramPacket(bufIn,
						bufIn.length);
				
				try {
					//every time check the message in hold back queue.
					System.out.println("holdBack: " + holdBack);
					if (holdBack.containsKey(sequenceNumber)) {
						process(holdBack.get(sequenceNumber));
						bufOut = Utils.toBytes(holdBack.get(sequenceNumber));
						//send here, but should know the IP and port
						port = getRMPort();
						pack = new DatagramPacket(bufOut, bufOut.length,
								serverIp, port);
						serverIp = pack.getAddress();
						dsOut.send(pack);
						System.out.println("process from hold back queue");
						holdBack.remove(sequenceNumber);
						sequenceNumber = sequenceNumber.add(new BigInteger("1"));
						break;
						
					}
					
					//listen for receive
					msIn.receive(pack);

					Message msg = (Message) Utils.fromBytes(bufIn);
					
					System.out.println("Received request " + msg.request + " seq# " + msg.sequenceNum + " from RM");
					
					//after receiving the message. if sequenceNumber is smaller than message's sequence number
					if (sequenceNumber.compareTo(msg.sequenceNum) < 0) {
						// Ask for re-send
//						byte bufOut[] = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
						for (int j = sequenceNumber.intValue(); j < msg.sequenceNum.intValue(); j++) {
							Message ackMsg = new Message();
							ackMsg.request = Message.REQUEST_NEGACK;
							ackMsg.sequenceNum = BigInteger.valueOf(j);
							ackMsg.replicaID = getReplicaId();
							bufOut = Utils.toBytes(ackMsg);
							port = getRMPort();
							pack = new DatagramPacket(bufOut, bufOut.length,
									pack.getAddress(), port);
							serverIp = pack.getAddress();
							dsOut.send(pack);
						
						}

						System.out.println("put the message in hold back queue");
						// Put message on hold-back queue
						holdBack.put(msg.sequenceNum, msg);

						
					}
					//sequence number is equal, then proceed. and add one on the seuqunceNumber
					else if (sequenceNumber.compareTo(msg.sequenceNum) == 0) {
						sequenceNumber = sequenceNumber.add(new BigInteger("1"));
						process(msg);
						System.out.println("get the message and process");
						//after processing, send 
						bufOut = Utils.toBytes(msg);
						port = getRMPort();
						pack = new DatagramPacket(bufOut, bufOut.length,
								pack.getAddress(), port);
						serverIp = pack.getAddress();
						
						System.out.println("serverIp " + serverIp + " port " + port + " result " + msg.result);
						
						dsOut.send(pack);
						
						
						
					}
					//sequenceNumber is larger than messge's sequence number
					//means this message has already proceed. then drop it
					else {
						System.out.println("drop the message");
						break;
					}

				} catch (Exception ex) {
					ex.printStackTrace();
				}
			}
		} catch (Exception ex) {
			ex.printStackTrace();

		} finally {
			if (msIn != null) {
				if (group != null) {
					try {
						msIn.leaveGroup(group);
					} catch (IOException ex) {
					}
				}
				msIn.close();
			}
			if (dsOut != null) {
				dsOut.close();
			}
		}
	}
	
	public String getReplicaId() {
		return replicaId;
	}

	public void setReplicaId(String replicaId) {
		this.replicaId = replicaId;
	}
}
